package com.I9lou.se.lurker.task.support;

import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.lucene.facet.taxonomy.TaxonomyWriter;
import org.apache.lucene.index.IndexWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.I9lou.se.lurker.facet.index.TaxonomyWriterFactory;
import com.I9lou.se.lurker.index.IndexWriterFactory;
import com.I9lou.se.lurker.task.Task;

public class IndexCommitTask implements Task {

	private static final Logger LOG = LoggerFactory.getLogger(IndexCommitTask.class);
	
	/**
	 * 索引提交间隔时间（单位：分钟），默认为30分钟提交一次
	 */
	private int commitInterval = 30;
	
	private IndexWriterFactory indexWriterFactory;
	
	private TaxonomyWriterFactory taxonomyWriterFactory;
	
	private ScheduledExecutorService scheduler;
	
	public void setIndexWriterFactory(IndexWriterFactory indexWriterFactory) {
		this.indexWriterFactory = indexWriterFactory;
	}

	public void setTaxonomyWriterFactory(TaxonomyWriterFactory taxonomyWriterFactory) {
		this.taxonomyWriterFactory = taxonomyWriterFactory;
	}
	
	public int getCommitInterval() {
		return commitInterval;
	}
	
	/**
	 * 调度器定时执行的时间间隔
	 * @param commitInterval
	 */
	public void setCommitInterval(int commitInterval) {
		this.commitInterval = commitInterval;
	}

	@Override
	public void init() {
		scheduler = Executors.newSingleThreadScheduledExecutor();
		long startTime = System.currentTimeMillis();
		scheduler.scheduleWithFixedDelay(new CommitTask(), commitInterval, commitInterval, TimeUnit.MINUTES);
		LOG.info( "Init IndexCommitTask success. cost: {}ms. Commit interval: {}min",
				  System.currentTimeMillis() - startTime,
				  commitInterval );
	}

	@Override
	public void destroy() {
		long startTime = System.currentTimeMillis();
		scheduler.shutdown();
		//打开对应factory destroy()运行的开关
		indexWriterFactory.getIndexWriterLatch().countDown();
		taxonomyWriterFactory.getTaxonomyWriterLatch().countDown();
		LOG.info( "Destory IndexCommitTask success. cost: {}ms.",
				  System.currentTimeMillis() - startTime);
	}

	@Override
	public String toString() {
		StringBuilder t_s = new StringBuilder("IndexCommitTask@{");
		t_s.append(" indexWriterFactory:[").append(indexWriterFactory).append("]");
		t_s.append(" taxonomyWriterFactory:[").append(taxonomyWriterFactory).append("]");
		return  t_s.toString();
	}

	private class CommitTask implements Runnable{

		@Override
		public void run() {
			try {
				//1.index writer commit
				for( String indexDirName : indexWriterFactory.getPool().keySet() ) {
					long startTime = System.currentTimeMillis();
					IndexWriter indexWriter = indexWriterFactory.getPool().get(indexDirName);
					try {
						indexWriter.commit();
						LOG.info("[Commit Task]{} commited! cost: {}ms", indexDirName, System.currentTimeMillis() - startTime);
					} catch(OutOfMemoryError e) {
						LOG.error("Encount OutOfMemoryError when commiting {}!", indexDirName);
						indexWriterFactory.close(indexDirName);
					}
				}
				
				//2.taxonomy writer commit
				long start = System.currentTimeMillis();
				Map<Integer,TaxonomyWriter> twMap = taxonomyWriterFactory.getTaxonomyWriter();
				for(TaxonomyWriter tw : twMap.values())
					tw.commit();
				//每次提交之后，都重建聚类reader
				taxonomyWriterFactory.recreateTaxonomyReader();
				LOG.info("[Commit Facet]: cost {}ms", (System.currentTimeMillis()-start));
			} catch(Exception e) {
				LOG.error("Encount unexcepted error when running taxonomy writer commit task!", e);
			}
		}
	}
	
}
